package com.yeziji.common.business.dataMessage.service.impl;

import cn.hutool.core.lang.UUID;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONWriter;
import com.mybatisflex.core.row.DbChain;
import com.yeziji.common.IServiceImpl;
import com.yeziji.common.business.dataMessage.base.DataMessage;
import com.yeziji.common.business.dataMessage.entity.DataMessageRecordEntity;
import com.yeziji.common.business.dataMessage.mapper.DataMessageRecordMapper;
import com.yeziji.common.business.dataMessage.service.DataMessageRecordService;
import com.yeziji.constant.DataMessageTopicEnum;
import com.yeziji.constant.DataSyncStatusEnum;
import com.yeziji.utils.RabbitMqUtils;
import com.yeziji.utils.ThreadPoolUtils;
import com.yeziji.utils.expansion.Opt2;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;

/**
 * 消息记录表 服务层实现。
 *
 * @author system
 * @since 2024-10-29
 */
@Slf4j
@Service
public class DataMessageRecordServiceImpl extends IServiceImpl<DataMessageRecordMapper, DataMessageRecordEntity> implements DataMessageRecordService {
    @Lazy
    @Resource
    private RabbitTemplate rabbitTemplate;

    @Override
    public void syncTo(DataMessageTopicEnum topicEnum, DataMessage dataMessage) {
        if (topicEnum == null || dataMessage == null) {
            return;
        }
        // 整合出需要发送的数据
        Long id = Opt2.nullElse(dataMessage.getId(), 0L);
        String dataJson = JSONObject.toJSONString(dataMessage);

        // 指定交换器和路由键值
        String exchangeKey = topicEnum.getExchange();
        String routeKey = topicEnum.getRouteKey();
        // 整合消息关联数据，可以交由 confirm 回调
        CorrelationData correlationData = new CorrelationData(UUID.fastUUID().toString());
        correlationData.setReturned(
                new ReturnedMessage(
                        new Message(dataJson.getBytes()),
                        0,
                        // reply txt 设置为业务的 id
                        id.toString(),
                        exchangeKey,
                        routeKey
                )
        );
        // 发送消息
        log.info("[{}] 发送同步云端: {}", correlationData.getId(), id);
        rabbitTemplate.convertAndSend(exchangeKey, routeKey, dataJson, correlationData);

        // 异步记录
        this.toSyncing(topicEnum, dataMessage, correlationData, null);
    }

    @Override
    public void sendTo(String exchangeKey, String routeKey, final DataMessage dataMessage, CorrelationData correlationData) {
        if (StrUtil.isBlank(exchangeKey) || StrUtil.isBlank(routeKey) || dataMessage == null) {
            return;
        }
        // 整合出需要发送的数据
        Long id = Opt2.nullElse(dataMessage.getId(), 0L);
        String dataJson = JSONObject.toJSONString(dataMessage);
        // 整合消息关联数据，可以交由 confirm 回调
        if (correlationData == null) {
            correlationData = new CorrelationData(UUID.fastUUID().toString());
            correlationData.setReturned(
                    new ReturnedMessage(
                            new Message(dataJson.getBytes()),
                            0,
                            // reply txt 设置为业务的 id
                            id.toString(),
                            exchangeKey,
                            routeKey
                    )
            );
        }
        rabbitTemplate.convertAndSend(exchangeKey, routeKey, dataJson, correlationData);

        // 判斷是什麽行爲: 如果是死信那么就直接发送, 反之如果指定了标识, 那么就要去除标识进行判断
        if (exchangeKey.startsWith("dead.letter")) {
            this.asyncRecord(DataMessageTopicEnum.DEAD_LETTER, dataMessage, correlationData, null, false);
        } else {
            String clientId = dataMessage.getClientId();
            // 先判断是否为操作主题
            DataMessageTopicEnum topicEnum = DataMessageTopicEnum.getByOperateExchange(exchangeKey);
            if (topicEnum == null) {
                // 如果不是先判断是否为客户端主题
                if (routeKey.startsWith(clientId)) {
                    topicEnum = DataMessageTopicEnum.getByExchangeAndRouteKey(exchangeKey, RabbitMqUtils.getAsRabbitNotPlatformKey(clientId, routeKey));
                } else {
                    topicEnum = DataMessageTopicEnum.getByExchangeAndRouteKey(exchangeKey, routeKey);
                }
            }
            if (topicEnum != null) {
                this.asyncRecord(topicEnum, dataMessage, correlationData, null, false);
            }
        }
    }

    @Override
    public DataMessageRecordEntity getByFlagId(String flagId) {
        if (flagId == null) {
            return null;
        }
        return this.queryChain().eq(DataMessageRecordEntity::getFlagId, flagId).one();
    }

    @Override
    public void retryFailed() {
        List<DataMessageRecordEntity> failedMessages = this.queryChain().eq(DataMessageRecordEntity::getStatus, DataSyncStatusEnum.FAILED.getCode()).list();
        if (CollectionUtils.isEmpty(failedMessages)) {
            log.info("🤩没有同步异常的数据哦！！！");
            return;
        }
        log.info("😨异常数据数量：{}", failedMessages.size());
        for (DataMessageRecordEntity failedMessage : failedMessages) {
            String topic = failedMessage.getTopic();
            if (StrUtil.isBlank(topic)) {
                continue;
            }
            // 必须存在 message
            String message = failedMessage.getMessage();
            if (message == null) {
                continue;
            }

            // 指定交换器和路由
            DataMessageTopicEnum topicEnum = DataMessageTopicEnum.getByName(topic);
            String exchange = topicEnum.getExchange();
            String routeKey = topicEnum.getRouteKey();

            // 将要发送的消息
            DataMessage dataMessage = JSONObject.parseObject(message, DataMessage.class);
            Long id = Opt2.nullElse(dataMessage.getId(), 0L);
            String dataJson = JSONObject.toJSONString(dataMessage);

            // 该值可能不存在
            String correlationDataStr = failedMessage.getCorrelationData();
            CorrelationData correlationData;
            if (StrUtil.isNotBlank(correlationDataStr)) {
                correlationData = JSONObject.parseObject(correlationDataStr, CorrelationData.class);
            } else {
                correlationData = new CorrelationData(UUID.fastUUID().toString());
                correlationData.setReturned(
                        new ReturnedMessage(
                                new Message(message.getBytes()),
                                0,
                                // reply txt 设置为业务的 id
                                id.toString(),
                                exchange,
                                routeKey
                        )
                );
            }

            // 发送消息
            log.info("[{}] 重试同步云端: {}", correlationData.getId(), id);
            rabbitTemplate.convertAndSend(exchange, routeKey, dataJson, correlationData);

            // 异步记录
            this.toSyncing(topicEnum, dataMessage, correlationData, failedMessage);
        }
    }

    /**
     * 异步记录发送主题的消息
     *
     * @param topicEnum       主题行为
     * @param dataMessage     发送的消息内容
     * @param correlationData 消息内容的关联数据
     * @param entity          消息记录实体类
     * @param toSync          是否更改同步狀態至同步中
     */
    private void asyncRecord(DataMessageTopicEnum topicEnum,
                             DataMessage dataMessage,
                             CorrelationData correlationData,
                             DataMessageRecordEntity entity,
                             boolean toSync) {
        // 异步保存消息记录
        ThreadPoolUtils.getCpuThreadPool().submit(() -> {
            String ip;
            try {
                ip = InetAddress.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                ip = "Unknown Host";
            }

            // save entity
            Long id = Opt2.nullElse(dataMessage.getId(), 0L);
            String dataJson = JSONObject.toJSONString(dataMessage);
            DataMessageRecordEntity recordEntity = DataMessageRecordEntity.builder()
                    .ip(ip)
                    .topic(topicEnum.toString())
                    .flagId(correlationData.getId())
                    .message(dataJson)
                    .correlationData(JSONObject.toJSONString(correlationData, JSONWriter.Feature.IgnoreErrorGetter))
                    .status(DataSyncStatusEnum.SYNCING.getCode())
                    .build();
            if (entity != null) {
                // 赋值 id 去更新
                recordEntity.setId(entity.getId());
            }
            this.saveOrUpdate(recordEntity);

            // update sync status
            if (id > 0L && toSync) {
                DbChain.table(dataMessage.getTable())
                        .set("sync_status", DataSyncStatusEnum.SYNCING.getCode())
                        // 将待同步数据发起同步
                        .where(String.format("id = %d and sync_status = %d", id, DataSyncStatusEnum.WAIT.getCode()))
                        .update();
            }
        });
    }


    /**
     * 异步记录发送主题的消息
     *
     * @param topicEnum       主题枚举
     * @param dataMessage     发送的消息内容
     * @param correlationData 消息内容的关联数据
     */
    private void toSyncing(DataMessageTopicEnum topicEnum, DataMessage dataMessage, CorrelationData correlationData, DataMessageRecordEntity entity) {
        this.asyncRecord(topicEnum, dataMessage, correlationData, entity, true);
    }
}
